package net.bwie.dwd;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class 草稿 {
    public static void main(String[] args) {
        EnvironmentSettings build = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tableEnv = TableEnvironment.create(build);

        Configuration configuration = tableEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        configuration.setString("execution.checkpointing.interval", "5 s");



        tableEnv.executeSql("CREATE TABLE dwd_retreat_behavior (\n" +
                //订单id
                "order_id STRING ,\n" +
                //  商品ID
                "product_id STRING ,\n" +
                // 订单支付时间
                "pay_time TIMESTAMP(3) ,\n" +
                //（订单总金额）
                "order_amount DOUBLE,\n"+

                // 退款id
                "refund_id STRING, \n"+
                // 退款申请时间
                "apply_time TIMESTAMP(3) ,\n" +
                //  退款场景:“退货退款”“已收货仅退款”“未收货退款”“未发货仅退款”
                "refund_scene STRING ,\n" +
                // 实际退款金额
                "refund_amount DOUBLE ,\n" +
                // (退款状态) /申请中/处理中/申请成功/申请关闭
                "refund_status STRING,\n"+

                //消费者行为id
                "behavior_id STRING ,\n" +
                //消费者id
                "user_id STRING ,\n" +
                //行为类型
                "behavior_type STRING \n" +
                ")WITH(\n" +
                "'connector' = 'kafka',\n" +
                "'topic' = 'dwd_retreat_behavior',\n" +
                "'properties.bootstrap.servers' = 'node101:9092',\n" +
                "'properties.group.id' = 'dwd_retreat_behavior-id078',\n" +
                "'scan.startup.mode' = 'earliest-offset',\n" +
                "'format' = 'json'\n" +
                ")");


        tableEnv.executeSql("SELECT *  FROM dwd_retreat_behavior WHERE refund_id IS NULL").print();

    }
}
